package emqx.test.consumer;

import org.eclipse.paho.client.mqttv3.*;

import java.nio.charset.StandardCharsets;

/**
 * 作为一个监听主题 消息回调的执行者 有一些方法
 * 需要约束(实现接口)
 */
public class EmqxDemoConsumer implements MqttCallbackExtended {
    /**
     * 异步监听线程 通知这个消费者 连接对象mqttClient已经建立完成
     */
    private MqttClient mqttClient;
    public EmqxDemoConsumer(MqttClient mqttClient) {
        this.mqttClient = mqttClient;
    }

    public EmqxDemoConsumer() {
    }

    @Override
    public void connectComplete(boolean reconnect, String serverURI) {
        System.out.println("连接对象mqttClient建立连接");
        //消费者 连接建立成功之后 订阅自己业务的主题 不同消费者订阅不同的主题
        //订阅监听主题 要通过mqttClient实现 /charging/order/submitC
        try {
            mqttClient.subscribe("/charging/order/submitC");
            mqttClient.subscribe("/charging/order/submitB");
            mqttClient.subscribe("/charging/order/submitA");
        } catch (MqttException e) {
            e.printStackTrace();
        }
    }
    /**
     * 异步监听线程 通知消费者,有一个你正在监听的主题中 产生了一条消息
     * 消费逻辑核心 通知消费者 消息来源主题(一个消费者不一定只监听一个主题)
     */
    @Override
    public void messageArrived(String topic, MqttMessage message) throws Exception {
        System.out.println("消息到达 topic:"+topic);
        byte[] payload = message.getPayload();
        //反序列化
        String msg=new String(payload, StandardCharsets.UTF_8);
        System.out.println("拿到消息 msg:"+msg);
    }
    /**
     * 异步监听线程 通知这个消费者 连接对象mqttClient断开了连接
     */
    @Override
    public void connectionLost(Throwable cause) {
        System.out.println("连接对象mqttClient断开连接,断开原因:"+cause.getMessage());
    }

    /**
     * 异步监听线程 通知消费者,某一条消息已经彻底消费结束 关系到qos级别 0 1 2
     * token表示的是哪条消息彻底消费投递结束
     */
    @Override
    public void deliveryComplete(IMqttDeliveryToken token) {
        try {
            System.out.println("消息投递结束:"+token.getMessage().getId());
        } catch (MqttException e) {
            throw new RuntimeException(e);
        }
    }
}
